// Copyright 2013 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.google.api.ads.dfa.matchtables;

import com.google.api.client.http.GenericUrl;
import com.google.api.client.http.HttpRequest;
import com.google.api.client.http.HttpResponse;
import com.google.api.client.util.DateTime;
import com.google.api.services.dfareporting.Dfareporting;
import com.google.api.services.dfareporting.model.DateRange;
import com.google.api.services.dfareporting.model.DimensionValue;
import com.google.api.services.dfareporting.model.File;
import com.google.api.services.dfareporting.model.Report;
import com.google.api.services.dfareporting.model.SortedDimension;

import au.com.bytecode.opencsv.CSVReader;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.Callable;

/**
 * Implementation of LookupObject that will generate match tables with paid search (dart search)
 * information.
 *
 * @author Moshe Kahan
 */
public class PaidSearchLookup implements LookupObject {

  private static final int MAX_ADVERTISERS = 500;
  // All requests to DFA Reporting will use retry logic.
  private static final int MAX_RETRIES = 3;
  // Polling for report completion settings. Will poll REPORT_POLL_MAX_RETRIES*MAX_RETRIES times.
  private static final int REPORT_POLL_INTERVAL_SECONDS = 30;
  private static final int REPORT_POLL_MAX_RETRIES = 400;
  private static Random randomGenerator = new Random();

  private Dfareporting dfaReporting;
  private TableSettings tableSettings;
  private long userProfileId;

  private Set<String> advertiserIdSet;
  private Set<String> keywordIdSet;
  private Set<String> timeSet;
  private int advertiserColumnIndex;
  private int keywordColumnIndex;
  private int timeColumnIndex;

  private Date dtFileStartDate;
  private Date dtFileEndDate;
  private final SimpleDateFormat timeFieldFormat;
  private final int dateStringStartIndex;
  private final int dateStringEndIndex;

  private List<File> paidSearchReports;

  /**
   * Constructor accepts DFA service/session objects needed to create Remotes to the
   * API.
   *
   * @param dfaReporting DFA Reporting object needed to create remotes to the DFA Reporting API
   * @param tableSettings match table application configuration used as a MatchTable object factory
   */
  PaidSearchLookup(Dfareporting dfaReporting, long userProfileId, TableSettings tableSettings) {
    this.dfaReporting = dfaReporting;
    this.tableSettings = tableSettings;
    this.userProfileId = userProfileId;

    // Full string would be "MM-dd-yyyy-hh:mm:ss".  We can ignore hours, minutes and seconds.
    String dateFormatString = "MM-dd-yyyy";
    dateStringStartIndex = 0;
    dateStringEndIndex = dateFormatString.length();
    timeFieldFormat = new SimpleDateFormat(dateFormatString);

    advertiserColumnIndex = -1;
    keywordColumnIndex = -1;
    timeColumnIndex = -1;

    advertiserIdSet = new HashSet<String>();
    keywordIdSet = new HashSet<String>();
    timeSet = new HashSet<String>();

    paidSearchReports = new ArrayList<File>();
  }

  /**
   * Called when LookupObjectManager is parsing the DT file header. Method stores index
   * of column of interest.
   * @param index in DT file columns
   * @param header name of this DT file field/column
   */
  public void addHeader(int index, String header) throws DataTransferFileException {
    if (LookupField.AdvertiserId.getFieldName().equals(header)) {
      advertiserColumnIndex = index;
    }
    if (LookupField.LegacyKeywordId.getFieldName().equals(header)) {
      keywordColumnIndex = index;
    }
    if (LookupObject.LookupField.Time.getFieldName().equals(header)) {
      timeColumnIndex = index;
    }
  }

  /**
   * Called to clear the DT header information, usually because you're processing a new file.
   */
  public void clearHeaders() {
    advertiserColumnIndex = -1;
    keywordColumnIndex = -1;
    timeColumnIndex = -1;
  }

  /**
   * Called to parse the DT row.
   *
   * @param values array of tokenized Strings of DT file line
   */
  public void addValues(String[] values) throws DataTransferFileException {
    if (keywordColumnIndex > -1) {
      String keywordId = values[keywordColumnIndex];
      if (keywordId != null && !keywordId.isEmpty() && !keywordId.equals("0")) {
        keywordIdSet.add(keywordId);
        // Add advertiser and time if keyword SV1 is present
        if (advertiserColumnIndex > -1) {
          String advertiserId = values[advertiserColumnIndex];
          if (advertiserId != null && !advertiserId.isEmpty() && !advertiserId.equals("0")) {
            advertiserIdSet.add(advertiserId);
          }
        }
        if (timeColumnIndex > -1) {
          try {
            // Parse only unique date substrings for faster parsing.
            String dateString = values[timeColumnIndex].substring(dateStringStartIndex,
                dateStringEndIndex);
            if (!timeSet.contains(dateString)) {
              timeSet.add(dateString);
              Date date = timeFieldFormat.parse(dateString);
              if (dtFileStartDate == null || date.before(dtFileStartDate)) {
                dtFileStartDate = date;
              }
              if (dtFileEndDate == null || date.after(dtFileEndDate)) {
                dtFileEndDate = date;
              }
            }
          } catch (ParseException e) {
            throw new DataTransferFileException("Error parsing Time field in DT file", e);
          }
        }
      }
    }
  }

  /**
   * This method needs to be called before the actual match table generation.
   * Retrieving paid search reports require a fair amount of server processing.
   * Client can call this method before all other match table generation
   * and then call generateMatchTable for PaidSearchLookup when the paid search
   * report is probably complete.
   */
  public void preGenerateMatchTable() throws DfaRemoteException {
    System.out.println("Pre-Generating rows for the dartsearch matchtable.");
    final DateRange dateRange = createDateRange();
    // Generate List of Advertiser IDs for report filtering.
    List<String> advertiserList = new ArrayList<String>(advertiserIdSet);
    final int size = advertiserIdSet.size();
    for (int i = 0, count = 0; i < size; i += MAX_ADVERTISERS, ++count) {
      BufferedReader reader = null;
      // Do MAX_ADVERTISERS in report at a time.
      int end = i + MAX_ADVERTISERS;
      if (end > size) {
        end = size;
      }
      final List<String> subAdvertiserList = advertiserList.subList(i, end);
      final int index = count;

      // Create the report.
      Report result = retry(new Callable<Report>() {
        public Report call() throws IOException {
          return createReport(subAdvertiserList, dateRange, index);
        }
      }, MAX_RETRIES);

      final Long reportId = result.getId();

      // Run the report.
      final File reportFile = retry(new Callable<File>() {
        public File call() throws IOException {
          return dfaReporting.reports().run(userProfileId, reportId).setSynchronous(false)
              .execute();
        }
      }, MAX_RETRIES);

      paidSearchReports.add(reportFile);
    }
  }

  /**
   * Called when parsing/aggregation is complete. Will lookup via API remote all unique DFA entries
   * in the DT file and pass data to MatchTable class.
   *
   * @param outputPath location of match table to be generated.
   */
  public void generateMatchTable(String outputPath)
      throws MatchTableFileException, DfaRemoteException {

    MatchTable dartSearchMatchTable = tableSettings.createMatchTable(MatchTable.Type.DARTSEARCH);
    if (dartSearchMatchTable != null) {
      dartSearchMatchTable.open(outputPath);
      Progress.printProgress(dartSearchMatchTable);
    }
    Spinner spinner = new Spinner();
    try {
      spinner.spin();
      for (File generatedReport : paidSearchReports) {
        final File report = generatedReport;
        // Get the report file handle.
        final File reportFile = retry(new Callable<File>() {
          public File call() throws DfaRemoteException, IOException {
            return waitForReport(report, REPORT_POLL_MAX_RETRIES, REPORT_POLL_INTERVAL_SECONDS);
          }
        }, MAX_RETRIES);

        // Download report from file handle URL.
        String url = reportFile.getUrls().getApiUrl();
        GenericUrl genericUrl = new GenericUrl(url);
        HttpRequest httpRequest = dfaReporting.getRequestFactory().buildGetRequest(genericUrl);
        HttpResponse response = httpRequest.execute();
        BufferedReader reader = new BufferedReader(new InputStreamReader(response.getContent(),
            "UTF-8"));
        parseReport(reader, dartSearchMatchTable);

        // Cleanup
        retry(new Callable<Boolean>() {
          public Boolean call() throws IOException {
            dfaReporting.reports().delete(userProfileId, report.getReportId())
            .execute();
            return true;
          }
        }, MAX_RETRIES);
      }
      spinner.stop();
      System.out.print("\r[100.0% of dartsearch lookups completed]");
    } catch (IOException e) {
      throw new DfaRemoteException("Exception encountered downloading Paid Search report", e);
    } finally {
      System.out.println();
      if (dartSearchMatchTable != null) {
        dartSearchMatchTable.close();
      }
    }
  }

  // Compute start and end times for report.
  // If Time column was absent from DT file, use last 7 days.
  private DateRange createDateRange() {
    DateRange dateRange = new DateRange();
    if (dtFileStartDate == null || dtFileEndDate == null) {
      dateRange.setRelativeDateRange("LAST_7_DAYS");
      System.out.println("Using 'LAST_7_DAYS' for our 'dartsearch date range'.");
    } else {
      SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
      dateRange.setStartDate(
          new DateTime(dateFormat.format(dtFileStartDate))).setEndDate(
              new DateTime(dateFormat.format(dtFileEndDate)));
      System.out.println("Using " + dateFormat.format(dtFileStartDate)
          + " for our 'dartsearch start date'.");
      System.out.println("Using " + dateFormat.format(dtFileEndDate)
          + " for our 'dartsearch end date'.");
    }
    return dateRange;
  }

  // Create the report with correct dimensions filtered by Advertiser IDs that have
  // associated keywords in DT file.
  private Report createReport(List<String> advertiserList, DateRange dateRange, int index)
      throws IOException {
    List<DimensionValue> filterList = new ArrayList<DimensionValue>(MAX_ADVERTISERS);
    for (String advertiserId : advertiserList) {
      DimensionValue advertiser = new DimensionValue().setDimensionName("dfa:advertiser")
          .setId(advertiserId);
      filterList.add(advertiser);
    }

    Report report = new Report();
    String reportPrefix = "MatchTable DS Report";
    if (dateRange.getStartDate() != null && dateRange.getEndDate() != null) {
      report.setName(String.format("%s (%s-%s:%d)",
          reportPrefix, dateRange.getStartDate(), dateRange.getEndDate(), index));
    } else if (dateRange.getRelativeDateRange() != null) {
      report.setName(String.format("%s (%s:%d)",
          reportPrefix, dateRange.getRelativeDateRange(), index));
    } else {
      report.setName(String.format("%s (%d)",
          reportPrefix, index));
    }

    report.setFileName("matchtable_ds_report_file");
    report.setType("STANDARD");

    Report.Criteria criteria = new Report.Criteria();
    criteria.setDateRange(dateRange);
    criteria.setDimensions(Arrays.asList(
        new SortedDimension().setName("dfa:adId"),
        new SortedDimension().setName("dfa:advertiserId"),
        new SortedDimension().setName("dfa:campaignId"),
        new SortedDimension().setName("dfa:paidSearchLegacyKeywordId"),
        new SortedDimension().setName("dfa:paidSearchKeywordId"),
        new SortedDimension().setName("dfa:paidSearchCampaign"),
        new SortedDimension().setName("dfa:paidSearchAdGroup"),
        new SortedDimension().setName("dfa:paidSearchBidStrategy"),
        new SortedDimension().setName("dfa:paidSearchLandingPageUrl"),
        new SortedDimension().setName("dfa:paidSearchKeyword"),
        new SortedDimension().setName("dfa:paidSearchMatchType")));
    // Need to add some metrics even though only interested in dimensions so rows will be returned.
    criteria.setMetricNames(Arrays.asList("dfa:paidSearchImpressions", "dfa:paidSearchClicks",
        "dfa:paidSearchActions"));
    criteria.setDimensionFilters(filterList);
    report.setCriteria(criteria);

    return dfaReporting.reports().insert(userProfileId, report)
        .execute();
  }

  // Polling for report to be completed.
  private File waitForReport(File file, int count, int seconds)
      throws DfaRemoteException, IOException {
    if (count < 0 || seconds < 0) {
      throw new IllegalArgumentException("count and seconds must be greater than zero");
    }
    if (file.getStatus().equals("PROCESSING")) {
      for (int i = 0; i <= count; ++i) {
        file = dfaReporting.reports().files().get(userProfileId, file.getReportId(),
            file.getId())
            .execute();
        if (!file.getStatus().equals("PROCESSING")) {
          break;
        }
        try {
          Thread.sleep(seconds * 1000);
        } catch (InterruptedException e) {
          e.printStackTrace(System.err);
          Thread.currentThread().interrupt();
          throw new DfaRemoteException("Received interrupt", e);
        }
      }
    }
    if (!file.getStatus().equals("REPORT_AVAILABLE")) {
      throw new DfaRemoteException("Paid Search Report Time Out even after "
          + (count * seconds) + " seconds. Report status: " + file.getStatus());
    }
    return file;
  }

  private void parseReport(BufferedReader reader, MatchTable dartSearchMatchTable)
      throws DfaRemoteException, IOException {
    // skip header lines
    String line;
    boolean headerLines = true;
    while ((line = reader.readLine()) != null) {
      if (line.startsWith("Report Fields")) {
        line = reader.readLine();
        // quick sanity check
        if (line == null || !line.startsWith("Ad ID")) {
          throw new DfaRemoteException("Malformed Paid Search Report");
        }
        break;
      }
    }

    // Read data rows
    if (line != null) {
      CSVReader csvReader = new CSVReader(reader);
      String[] fields = null;
      if (dartSearchMatchTable != null) {
        while ((fields = csvReader.readNext()) != null) {
          // Only write rows whose keyword IDs match what is in the DT file.
          if (keywordIdSet.contains(fields[3])) {
            dartSearchMatchTable.write(fields[0], fields[1], fields[2], fields[3], fields[4],
                fields[5], fields[6], fields[7], fields[8], fields[9], fields[10]);
          }
        }
      }
    }
  }

  // Retry method with random sleeps.
  private static <T> T retry(Callable<T> callable, int retries) throws DfaRemoteException {
    Exception cause = null;
    int attempts = 0;
    do {
      ++attempts;
      try {
        return callable.call();
      } catch (Exception e) {
        cause = e;
        System.err.println("ERROR! Attempt #" + attempts + " with exception: " + e);
        e.printStackTrace(System.err);
      }
      try {
        // Sleep for 1 to 5 seconds.
        Thread.sleep(randomGenerator.nextInt(5) * 1000 + 1000); 
      } catch (InterruptedException e) {
        e.printStackTrace(System.err);
        Thread.currentThread().interrupt();
        throw new DfaRemoteException("Received interrupt", e);
      }
    } while (attempts < retries);
    throw new DfaRemoteException("Error in using DFA Remote service", cause);
  }
}
